package org.finesys.common.hazelcast.config;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.finesys.common.hazelcast.core.base.DistributedBaseFactory;
import org.finesys.common.hazelcast.core.cache.DefaultCacheManager;
import org.finesys.common.hazelcast.core.cache.DistributedCacheProvider;
import org.finesys.common.hazelcast.core.cache.HazelcastCacheProvider;
import org.finesys.common.hazelcast.core.mq.AbstractGroupMessageListener;
import org.finesys.common.hazelcast.core.mq.AbstractTopicMessageListener;
import org.finesys.common.hazelcast.core.mq.HazelcastMQTemplate;
import org.finesys.common.hazelcast.core.mq.HazelcastMessageInterceptor;
import org.finesys.common.hazelcast.core.mq.HazelcastMessageListenerProvider;
import org.finesys.common.hazelcast.core.support.HazecastBaseFactory;
import org.finesys.common.hazelcast.core.support.HazelcastClientFactory;
import org.finesys.common.hazelcast.core.support.HazelcastProperties;
import org.finesys.common.hazelcast.core.support.HazelcastServerInstance;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import com.hazelcast.core.HazelcastInstance;

@AutoConfiguration
@EnableConfigurationProperties(HazelcastProperties.class)
public class HazelcastAutoConfiguration {

    @Bean
    DistributedBaseFactory distributedFactory(HazelcastInstance instance) {
        return new HazecastBaseFactory(instance);
    }

    /**
     * 客户端实例信息创建
     */
    @Bean
    HazelcastClientFactory hazelcastInstance(HazelcastProperties hazelcastProperties, ObjectProvider<HazelcastServerInstance> serverInstances) {
        return new HazelcastClientFactory(hazelcastProperties, serverInstances.getIfAvailable());
    }

    /**
     * Hazelcast节点配置
     *
     * @param hazelcastProperties hazelcast节点相关配置
     * @return hazelcast实例
     */
    @ConditionalOnProperty(value = HazelcastProperties.SERVER, matchIfMissing = true, havingValue = "true")
    @Bean
    HazelcastServerInstance hazelcastServerInstance(HazelcastProperties hazelcastProperties) {
        return new HazelcastServerInstance(hazelcastProperties);
    }

    @Bean
    DistributedCacheProvider distributedCacheProvider(HazelcastInstance instance) {
        return new HazelcastCacheProvider(instance);
    }

    @Bean
    DefaultCacheManager defaultCacheManager(DistributedCacheProvider cacheProvider) {
        return new DefaultCacheManager(cacheProvider);
    }


    @Bean
    ExecutorService executorService(HazelcastProperties hazelcastProperties) {
        int corePoolSize = hazelcastProperties.getCorePoolSize();
        int maximumPoolSize = hazelcastProperties.getMaximumPoolSize();
        long keepAliveTime = hazelcastProperties.getKeepAliveTime();
        int capacity = hazelcastProperties.getCapacity();
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue;
        workQueue = corePoolSize <= 0 ? new SynchronousQueue<>() : new LinkedBlockingQueue<>(capacity);
        ThreadFactory threadFactory = new CustomizableThreadFactory("hazelcastThreadFactory-");
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        return threadPoolExecutor;
    }


    @Bean
    HazelcastMQTemplate hazelcastMQTemplate(HazelcastInstance hazelcastInstance,
                                            List<HazelcastMessageInterceptor> interceptors) {
        return new HazelcastMQTemplate(hazelcastInstance, interceptors);
    }

    @Bean
    HazelcastMessageListenerProvider hazelcastMessageListenerProvider(
            HazelcastInstance hazelcastInstance,
            ExecutorService executorService,
            List<AbstractTopicMessageListener<?>> topicMessageListeners,
            List<AbstractGroupMessageListener<?>> groupMessageListeners) {
        HazelcastMessageListenerProvider hazelcastMessageListenerProvider = new HazelcastMessageListenerProvider(
                hazelcastInstance, executorService);
        topicMessageListeners.forEach(hazelcastMessageListenerProvider::register);
        groupMessageListeners.forEach(hazelcastMessageListenerProvider::register);
        return hazelcastMessageListenerProvider;
    }
}
